package com.bfxy.disruptor.common.consume;

import java.util.concurrent.atomic.AtomicBoolean;

import com.bfxy.disruptor.common.event.EventReleaseAware;
import com.bfxy.disruptor.common.event.EventReleaser;
import com.bfxy.disruptor.common.LifecycleAware;
import com.bfxy.disruptor.common.RingBuffer;
import com.bfxy.disruptor.common.WorkHandler;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.exception.AlertException;
import com.bfxy.disruptor.common.exception.ExceptionHandler;
import com.bfxy.disruptor.common.producer.Productor;
//多消费者模式使用WorkProcessor
public final class MutilConsumer<T> implements EventConsumer {
	private final AtomicBoolean running = new AtomicBoolean(false);
	private final Sequence sequence = new Sequence(Productor.INITIAL_CURSOR_VALUE);
	private final RingBuffer<T> ringBuffer;
	private final SequenceBarrier sequenceBarrier;
	private final WorkHandler<? super T> workHandler;
	private final ExceptionHandler<? super T> exceptionHandler;
	private final Sequence workSequence;

	private final EventReleaser eventReleaser = new EventReleaser() {
		@Override
		public void release() {
			sequence.set(Long.MAX_VALUE);
		}
	};

	public MutilConsumer(final RingBuffer<T> ringBuffer, final SequenceBarrier sequenceBarrier,
                        final WorkHandler<? super T> workHandler, final ExceptionHandler<? super T> exceptionHandler,
                        final Sequence workSequence) {
		this.ringBuffer = ringBuffer;
		this.sequenceBarrier = sequenceBarrier;
		this.workHandler = workHandler;
		this.exceptionHandler = exceptionHandler;
		this.workSequence = workSequence;

		if (this.workHandler instanceof EventReleaseAware) {
			((EventReleaseAware) this.workHandler).setEventReleaser(eventReleaser);
		}
	}

	@Override
	public Sequence getSequence() {
		return sequence;
	}

	@Override
	public void halt() {
		running.set(false);
		sequenceBarrier.alert();
	}

	@Override
	public boolean isRunning() {
		return running.get();
	}

	@Override
	public void run() {
		if (!running.compareAndSet(false, true)) {
			throw new IllegalStateException("Thread is already running");
		}
		sequenceBarrier.clearAlert();

		notifyStart();

		boolean processedSequence = true;
		long cachedAvailableSequence = Long.MIN_VALUE;
		long nextSequence = sequence.get();
		T event = null;
		while (true) {
			try {
				// if previous sequence was processed - fetch the next sequence and set
				// that we have successfully processed the previous sequence
				// typically, this will be true
				// this prevents the sequence getting too far forward if an exception
				// is thrown from the WorkHandler
				if (processedSequence) {
					processedSequence = false;
					do {
						nextSequence = workSequence.get() + 1L;
						sequence.set(nextSequence - 1L);
					} while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
				}

				if (cachedAvailableSequence >= nextSequence) {
					event = ringBuffer.get(nextSequence);
					workHandler.onEvent(event);// 调用消费者
					processedSequence = true;
				} else {
					cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
				}
			} catch (final AlertException ex) {
				if (!running.get()) {
					break;
				}
			} catch (final Throwable ex) {
				// handle, mark as processed, unless the exception handler threw an exception
				exceptionHandler.handleEventException(ex, nextSequence, event);
				processedSequence = true;
			}
		}

		notifyShutdown();

		running.set(false);
	}

	private void notifyStart() {
		if (workHandler instanceof LifecycleAware) {
			try {
				((LifecycleAware) workHandler).onStart();
			} catch (final Throwable ex) {
				exceptionHandler.handleOnStartException(ex);
			}
		}
	}

	private void notifyShutdown() {
		if (workHandler instanceof LifecycleAware) {
			try {
				((LifecycleAware) workHandler).onShutdown();
			} catch (final Throwable ex) {
				exceptionHandler.handleOnShutdownException(ex);
			}
		}
	}
}
